package com.qidian.by.search.service.impl;

import com.qidian.by.search.dto.SearchItem;
import com.qidian.by.search.service.SearchReceive;
import com.qidian.by.search.service.SearchItemService;
import com.qidian.by.util.pojo.EsResult;
import com.qidian.by.util.pool.ThreadTools;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.Objects;

/**
 * @Author: CoolPro
 * @Description:
 * @Time: 2024-05-17 20:05
 **/
@Service
public class SearchReceiveImpl implements SearchReceive {
    @Resource
    private SearchItemService searchItemService;
    @Resource
    private StringRedisTemplate stringRedisTemplate;


    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
//        统一的接收事件
        SearchItem item = null;
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            //拿到消息
            item = EsResult.json2object(new String(message.getBody(), StandardCharsets.UTF_8), SearchItem.class);
            //只要拿到了消息，就立刻确认
            channel.basicAck(deliveryTag, true);
            if (Objects.nonNull(item)) {
                processMessages(item);
            }
        } catch (Exception e) {
            if (Objects.isNull(item)) {
                channel.basicAck(deliveryTag, false);
                handleException(new String(message.getBody(), StandardCharsets.UTF_8));
            }
        }
    }

    ThreadLocal<Integer> count = new ThreadLocal<>();

    public void handleException(String msg) {
        //处理消息发送失败的异常--默认策略是重试5次，如果还是失败，就把他的值记录在redis中
        ThreadTools.redisDelayed(2, () -> {
            handleException(msg);
            count.set(count.get() + 1);
            if (count.get() > 5) {
                stringRedisTemplate.opsForValue().set("SEARCH_SYNC_FAL", msg);
                return;
            } else {
                processMessages(EsResult.json2object(msg, SearchItem.class));
            }
        });
    }

    public void processMessages(SearchItem item) {
        switch (item.getOption()) {
            case "INSERT": {
                searchItemService.saveSearchItem(item);
            }
            break;
            case "UPDATE": {
//                        更新
            }
            break;
            case "DELETE": {
//                        删除
            }
            break;
        }
    }
}
